package com.xiehua.task.job.bucket;

import com.xiehua.task.job.AbsJobOperation;
import com.xiehua.task.job.pool.DelayJob;
import com.xiehua.task.job.pool.DelayJobPool;
import com.xiehua.task.job.pool.Job;
import redis.clients.jedis.Jedis;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import static com.xiehua.task.job.pool.Job.JobStatus.delay;

public class DelayBucket extends AbsJobOperation<DelayJob> {

    private DelayJobPool delayJobPool;

    public DelayBucket(Jedis jedis) {
        super(jedis);
        delayJobPool = new DelayJobPool(jedis);
    }

    /***
     * 添加延迟任务
     * return jobId
     * @param job
     **/
    @Override
    public DelayJob addJob(DelayJob job) {
        //1.put current job to fixed job pool
        delayJobPool.push(job);
        //3.job bucket add record
       machiningJob(job);
        return job;
    }

    /***
     * 更新job
     * @param job
     **/
    @Override
    public void updateJob(DelayJob job) {
        delayJobPool.updateJob(job);
    }

    /***
     * 查询job
     * @param jobId
     **/
    @Override
    public DelayJob queryJob(String jobId) {
        return delayJobPool.get(jobId);
    }
    /***
     * 查询等待执行的job
     * **/
    @Override
    public List<DelayJob> queryCurrentJob() {
        Set<String> topics = new TopicBucket(jedis).queryTopics();
        final Long min = LocalDateTime.now().minusHours(1).toEpochSecond(ZoneOffset.of("+8"));
        final Long max = LocalDateTime.now().toEpochSecond(ZoneOffset.of("+8"));
        return topics.stream().map(s -> {
            String zsetKey = JOB_BUCKET_PIX + s;
            Set<String> jobIds = jedis.zrangeByScore(zsetKey, min, max);
            return jobIds;
        }).flatMap(t -> t.stream()).map(m -> queryJob(m)).filter(t -> t != null && delay.equals(t.getStatus()) && Job.JobType.delay.equals(t.getType())).collect(Collectors.toList());
    }

}
